package com.ams.server.replication;

import java.io.IOException;
import java.util.HashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ams.protocol.rtmp.RtmpConnection;
import com.ams.protocol.rtmp.RtmpException;
import com.ams.protocol.rtmp.amf.AmfException;
import com.ams.protocol.rtmp.amf.AmfValue;
import com.ams.protocol.rtmp.message.RtmpMessage;
import com.ams.protocol.rtmp.message.RtmpMessageAck;
import com.ams.protocol.rtmp.message.RtmpMessageCommand;
import com.ams.protocol.rtmp.net.NetStream;
import com.ams.protocol.rtmp.net.NetStreamManager;
import com.ams.protocol.rtmp.net.StreamPublisherManager;
import com.ams.protocol.rtmp.net.StreamPublisher;
import com.ams.server.handler.RtmpHandler;

public class ReplPublisher {
    final private Logger logger = LoggerFactory.getLogger(RtmpHandler.class);
    final static int PING_SUBSCRIBER_INTERVAL = 1000;

    private RtmpConnection rtmp;
    private NetStreamManager streamManager;
    private HashMap<String, NetStream> publishStreams = new HashMap<String, NetStream>();

    private long lastPingTime = 0;;

    public ReplPublisher(RtmpConnection rtmp) {
        this.rtmp = rtmp;
        this.streamManager = new NetStreamManager();
    }

    private void sendPublishCommand(NetStream stream, String publishName) {
        try {
            AmfValue[] args = AmfValue.array(null, publishName, "live");
            RtmpMessage message = new RtmpMessageCommand("publish", 1, args);
            stream.writeMessage(message);
        } catch (IOException e) {
        }
    }

    private void sendCloseStreamCommand(NetStream stream) throws IOException {
        AmfValue[] args = { new AmfValue(null) };
        RtmpMessage message = new RtmpMessageCommand("closeStream", 0, args);
        stream.writeMessage(message);
    }
    
    private void startPublishStream(String publishName) {
        // is publishing
        if (publishStreams.containsKey(publishName)) {
            logger.info("the stream is publishing: {}", publishName);
            return;
        }
        StreamPublisher publisher = (StreamPublisher) StreamPublisherManager
                .getPublisher(publishName);
        if (publisher == null) {
            logger.info("not found publisher: {}", publishName);
            return;
        }
        NetStream stream = streamManager.createStream(rtmp);
        publisher.addSubscriber(stream);
        // this subscriber will publish stream to remote subscriber
        publishStreams.put(publishName, stream);
        
        // send publish command
        sendPublishCommand(stream, publishName);
    }

    public void pingSubscriber() throws IOException {
        // ping subscriber every 1s to keep subscriber alive.
        long now = System.currentTimeMillis();
        if (now - lastPingTime > PING_SUBSCRIBER_INTERVAL) {
            rtmp.writeProtocolControlMessage(new RtmpMessageAck(1));
            lastPingTime = now;
        }

    }

    public void tryToCloseSubscribeStreams() throws IOException {
        // try to close the streams that have been closed
        for (String publishName : publishStreams.keySet()) {
            NetStream subscriberStream = publishStreams.get(publishName);
            if (StreamPublisherManager.getPublisher(publishName) == null) {
                logger.info("close publish stream: {}", publishName);
                sendCloseStreamCommand(subscriberStream);
                publishStreams.remove(publishName);
            }
        }
    }

    public void acceptSubscription() throws IOException, AmfException,
            RtmpException {
        if (!rtmp.readRtmpMessage())
            return;
        RtmpMessage message = rtmp.getCurrentMessage();
        if (message.getType() == RtmpMessage.MESSAGE_AMF0_COMMAND) {
            RtmpMessageCommand command = (RtmpMessageCommand) message;
            if ("subscribe".equals(command.getName())) {
                AmfValue[] args = command.getArgs();
                // subscriber want to play this stream
                String streamName = args[1].string();
                logger.info("received subscription, start to publish stream: {}", streamName);
                startPublishStream(streamName);
            }
        }
    }

}